// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <cstdint>
#include <string>
#include <vector>

#include <nebula/array/data.h>
#include <nebula/compute/api_vector.h>
#include <nebula/compute/exec.h>
#include <nebula/compute/function.h>
#include <nebula/compute/kernel.h>
#include <nebula/compute/kernels/codegen_internal.h>

namespace nebula::compute::internal {

    using FilterState = OptionsWrapper<FilterOptions>;
    using TakeState = OptionsWrapper<TakeOptions>;

    struct SelectionKernelData {
        InputType value_type;
        InputType selection_type;
        ArrayKernelExec exec;
    };

    void register_selection_function(const std::string &name, FunctionDoc doc,
                                   VectorKernel base_kernel,
                                   std::vector<SelectionKernelData> &&kernels,
                                   const FunctionOptions *default_options,
                                   FunctionRegistry *registry);

    /// \brief Callback type for VisitPlainxREEFilterOutputSegments.
    ///
    /// position is the logical position in the values array relative to its offset.
    ///
    /// segment_length is the number of elements that should be emitted.
    ///
    /// filter_valid is true if the filter run value is non-NULL. This value can
    /// only be false if null_selection is NullSelectionBehavior::EMIT_NULL. For
    /// NullSelectionBehavior::DROP, NULL values from the filter are simply skipped.
    ///
    /// Return true if iteration should continue, false if iteration should stop.
    using EmitREEFilterSegment =
            std::function<bool(int64_t position, int64_t segment_length, bool filter_valid)>;

    void VisitPlainxREEFilterOutputSegments(
            const ArraySpan &filter, bool filter_may_have_nulls,
            FilterOptions::NullSelectionBehavior null_selection,
            const EmitREEFilterSegment &emit_segment);

    turbo::Status PrimitiveFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status ListFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status LargeListFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status ListViewFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status LargeListViewFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status FSLFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status DenseUnionFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status MapFilterExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status VarBinaryTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status LargeVarBinaryTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status FixedWidthTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status ListTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status LargeListTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status ListViewTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status LargeListViewTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status FSLTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status DenseUnionTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status SparseUnionTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status StructTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

    turbo::Status MapTakeExec(KernelContext *, const ExecSpan &, ExecResult *);

}  // namespace nebula::compute::internal
